/*
 * Copyright 2011-2012 by The Regents of the University of California Licensed
 * under the Apache License, Version 2.0 (the "License"); you may not use this
 * file except in compliance with the License. you may obtain a copy of the
 * License from
 * 
 * http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
package edu.ucsd.db.bassarisk.storage.common;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.storage.common.buffercache.BufferCache;
import edu.uci.ics.hyracks.storage.common.buffercache.ClockPageReplacementStrategy;
import edu.uci.ics.hyracks.storage.common.buffercache.HeapBufferAllocator;
import edu.uci.ics.hyracks.storage.common.buffercache.IBufferCache;
import edu.uci.ics.hyracks.storage.common.buffercache.ICacheMemoryAllocator;
import edu.uci.ics.hyracks.storage.common.buffercache.IPageReplacementStrategy;
import edu.uci.ics.hyracks.storage.common.file.IFileMapManager;
import edu.uci.ics.hyracks.storage.common.file.IFileMapProvider;
import edu.uci.ics.hyracks.storage.common.smi.TransientFileMapManager;
import edu.ucsd.db.bassarisk.storage.loading.CabinetLoader;
import edu.ucsd.db.bassarisk.storage.loading.HyracksBTreeLexiconLoader;
import edu.ucsd.db.bassarisk.storage.reading.CabinetReader;
import edu.ucsd.db.bassarisk.storage.reading.HyracksBTreeLexiconReader;

public class Cabinet {

  private static final int PAGE_SIZE = 32768;
  private static final int NUM_PAGES = 4096;
  private static final int MAX_OPEN_FILES = 16;

  private final IHyracksTaskContext ctx;
  private final IFileMapProvider fileMap;
  private final IBufferCache bufferCache;

  private final String lexFileName;
  private final String cabFileName;

  private CabinetLoader cabLoader;
  private CabinetReader cabReader;

  public Cabinet(final IHyracksTaskContext ctx, final String fileNameBase) {
    this.lexFileName = fileNameBase + ".lex";
    this.cabFileName = fileNameBase + ".cab";

    this.ctx = ctx;
    this.fileMap = new TransientFileMapManager();

    // initialize buffer cache
    ICacheMemoryAllocator allocator = new HeapBufferAllocator();
    IPageReplacementStrategy prs = new ClockPageReplacementStrategy();
    this.bufferCache =
        new BufferCache(this.ctx.getIOManager(), allocator, prs,
                        (IFileMapManager) this.fileMap, PAGE_SIZE, NUM_PAGES,
                        MAX_OPEN_FILES);
  }

  public void close() {
    if (this.cabLoader != null) {
      this.cabLoader.close();
    }
    if (this.cabReader != null) {
      this.cabReader.close();
    }
    this.bufferCache.close();
  }

  public CabinetLoader makeLoader() {
    if (this.cabLoader == null) {
      try {
        HyracksBTreeLexiconLoader lexLoader = new HyracksBTreeLexiconLoader();
        lexLoader.initialize(this.bufferCache, this.lexFileName, this.fileMap);
        this.cabLoader =
            new CabinetLoader(this.bufferCache, this.cabFileName, this.fileMap,
                              lexLoader);
      } catch (HyracksDataException e) {
        throw new RuntimeException(e);
      }
    }
    return this.cabLoader;
  }

  public CabinetReader makeReader() {
    if (this.cabReader == null) {
      try {
        HyracksBTreeLexiconReader lexReader = new HyracksBTreeLexiconReader();
        lexReader.initialize(this.bufferCache, this.lexFileName, this.fileMap);
        this.cabReader =
            new CabinetReader(this.bufferCache, this.cabFileName, this.fileMap,
                              lexReader);
      } catch (HyracksDataException e) {
        throw new RuntimeException(e);
      }
    }
    return this.cabReader;
  }
}
